package cn.jly.flink.cep;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.List;
import java.util.Map;


/**
 * 复杂事件处理 Complex Event Processing（CEP）是 Flink 提供的一个非常亮眼的功能
 * <p>
 * 说到底，Flink 的 CEP 到底解决了什么样的问题呢？
 * <p>
 * 比如，我们需要在大量的订单交易中发现那些虚假交易，在网站的访问日志中寻找那些使用脚本或者工具“爆破”登录的用户，或者在快递运输中发现那些滞留很久没有签收的包裹等。
 * <p>
 * Flink CEP 的程序结构主要分为两个步骤：
 * - 定义模式
 * - 匹配结果
 * <p>
 * 案例
 * 我们模拟电商网站用户搜索的数据来作为数据的输入源，然后查找其中重复搜索某一个商品的人，并且发送一条告警消息
 *
 * @author lanyangji
 * @create 2020-09-03 17:32
 */
public class CepApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 数据源
        DataStreamSource<BrowseRecord> datas = env.fromElements(
                //浏览记录
                new BrowseRecord("Marry", "外套", 1L),
                new BrowseRecord("Marry", "帽子", 1L),
                new BrowseRecord("Marry", "帽子", 2L),
                new BrowseRecord("Marry", "帽子", 3L),
                new BrowseRecord("Ming", "衣服", 1L),
                new BrowseRecord("Marry", "鞋子", 1L),
                new BrowseRecord("Marry", "鞋子", 2L),
                new BrowseRecord("LiLei", "帽子", 1L),
                new BrowseRecord("LiLei", "帽子", 2L),
                new BrowseRecord("LiLei", "帽子", 3L)
        );

        // 定义Pattern,寻找连续搜索帽子的用户
        Pattern<BrowseRecord, BrowseRecord> pattern = Pattern.<BrowseRecord>begin("start")
                .where(new SimpleCondition<BrowseRecord>() {
                    @Override
                    public boolean filter(BrowseRecord value) throws Exception {
                        return "帽子".equals(value.productName);
                    }
                })
                //.timesOrMore(3)
                .next("middle")
                .where(new SimpleCondition<BrowseRecord>() {
                    @Override
                    public boolean filter(BrowseRecord value) throws Exception {
                        return "帽子".equals(value.productName);
                    }
                });

        PatternStream<BrowseRecord> patternStream = CEP.pattern(
                datas.keyBy((KeySelector<BrowseRecord, String>) value -> value.customerName),
                pattern
        );

        patternStream.select(new PatternSelectFunction<BrowseRecord, String>() {
            @Override
            public String select(Map<String, List<BrowseRecord>> pattern) throws Exception {
                List<BrowseRecord> browseRecords = pattern.get("middle");
                return browseRecords.get(0).customerName + ": " + browseRecords.get(0).productName + " 连续搜索了两次帽子";
            }
        })
                .printToErr();

        env.execute("CepApp");
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class BrowseRecord {
        private String customerName;
        private String productName;
        private Long browseTime;
    }
}
